package org.example.verify;

import com.antgroup.geaflow.common.tuple.Tuple;
import org.example.csv.CSVData;
import org.example.util.CSVUtils;
import org.example.util.FileUtils;
import org.junit.Test;

import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class Case3Verify {

    public static final String basicPath = "D:\\Development\\GeaFlowCompetition\\sf1\\snapshot";

    public static final String outputPath = "D:\\Development\\GeaFlowCompetition\\output";

    public static final String outputName = "result3-sf1-check.csv";

    Map<String, List<Tuple<String, Double>>> inMap = new HashMap<>();

    Map<String, List<Tuple<String, Double>>> outMap = new HashMap<>();

    Map<String, Double> res = new HashMap<>();

    @Test
    public void verify() {
        Path path = FileUtils.pathJoin(basicPath, "AccountTransferAccount.csv");
        CSVData csvData = CSVUtils.read(path);

        List<String> fromList = csvData.getByColumn("fromId");
        List<String> toList = csvData.getByColumn("toId");
        List<String> amountList = csvData.getByColumn("amount");

        for (int i = 0; i < fromList.size(); i++) {
            String from = fromList.get(i);
            String to = toList.get(i);
            Double amount = Double.parseDouble(amountList.get(i));
            List<Tuple<String, Double>> outList = outMap.getOrDefault(from, new ArrayList<>());
            outList.add(Tuple.of(to, amount));
            outMap.put(from, outList);

            List<Tuple<String, Double>> inList = inMap.getOrDefault(to, new ArrayList<>());
            inList.add(Tuple.of(from, amount));
            inMap.put(to, inList);
        }

        for (String key : outMap.keySet()) {
            if (inMap.containsKey(key)) {
                double inSum = 0.0, outSum = 0.0;
                List<Tuple<String, Double>> outs = outMap.get(key);
                for (Tuple<String, Double> out : outs) {
                    outSum += out.f1;
                }

                List<Tuple<String, Double>> ins = inMap.get(key);
                for (Tuple<String, Double> in : ins) {
                    inSum += in.f1;
                }

                res.put(key, inSum / outSum);
            }
        }

        CSVData writer = new CSVData(Arrays.asList("id", "value"));
        List<Long> ids = new ArrayList<>();
        for (String key : res.keySet()) {
            ids.add(Long.valueOf(key));
        }
        Collections.sort(ids);
        for (Long id : ids) {
            String key = String.valueOf(id);
            writer.addLine(new String[]{key, String.format("%.2f", res.get(key))});
        }
        CSVUtils.write(FileUtils.pathJoin(outputPath, outputName), writer);
    }
}
